package com.cu.kafka.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Service
public class KafkaProducerDemo {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void send(String topic, String message) {
         kafkaTemplate.send(topic, message).addCallback(success -> {
             if (success != null){
                 System.out.println("Message sent successfully.");
             }
         },failure  ->{
             System.out.println("发送消息失败：" + failure.getMessage());
         });
    }

    public void send(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message).addCallback(success -> {
            if (success != null){
                System.out.println("Message sent successfully.");
            }
        },failure  ->{
            System.out.println("发送消息失败：" + failure.getMessage());
        });;
    }

    public void send(String topic, Integer partition, String key, String message) {
        kafkaTemplate.send(topic, partition, key, message).addCallback(success -> {
            if (success != null){
                System.out.println("Message sent successfully.");
            }
        },failure  ->{
            System.out.println("发送消息失败：" + failure.getMessage());
        });;
    }
}
